package com.ccb.channel.atm;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.alibaba.fastjson.JSON;
import com.ccb.util.Record;
import com.ccb.util.RecordType;
import com.ccb.util.StandardRecord;

public class AtmReducer extends Reducer<Text, Text, Text, NullWritable> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(AtmReducer.class);

	private MultipleOutputs<Text, NullWritable> mos = null;

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - start");
		}

		mos = new MultipleOutputs<Text, NullWritable>(context);

		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - end");
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - start");
		}

		mos.close();

		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - end");
		}
	}

	public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("reduce(Text, Iterable<Text>, Context) - start");
		}

		Map<String, Record> reqMap = new HashMap<>();
		Map<String, Record> rspMap = new HashMap<>();
		StandardRecord standardRecord = new StandardRecord();
		for (Text val : values) {
			Record rd = JSON.parseObject(val.toString(), Record.class);
			if (rd.getRecordType().equals(RecordType.REQUEST)) {
				if (rspMap.containsKey(rd.getUid())) {
					Record rspRd = rspMap.get(rd.getUid());
					standardRecord.setUid(rd.getUid());
					standardRecord.setServiceId(rd.getServiceId());

					standardRecord.setRequestData(rd.getData());
					standardRecord.setRequest(rd.getOriginalData());
					standardRecord.setTime(rd.getTime());

					standardRecord.setResponseData(rspRd.getData());
					standardRecord.setResponse(rspRd.getOriginalData());
					standardRecord.setResponseCode(rspRd.getResponseCode());

					standardRecord.setRecordType(RecordType.BOTH);

					rspMap.remove(rd.getUid());

					mos.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get(),  rd.getServiceId());
					// context.write(new Text(JSON.toJSONString(standardRecord)), null);
				} else {
					reqMap.put(rd.getUid(), rd);
				}
			}else if (rd.getRecordType().equals(RecordType.RESPONSE)) {
				if (reqMap.containsKey(rd.getUid())) {
					Record reqRd = reqMap.get(rd.getUid());
					standardRecord.setUid(rd.getUid());
					standardRecord.setServiceId(rd.getServiceId());

					standardRecord.setRequestData(reqRd.getData());
					standardRecord.setRequest(reqRd.getOriginalData());
					standardRecord.setTime(reqRd.getTime());

					standardRecord.setResponseCode(rd.getResponseCode());
					;
					standardRecord.setResponse(rd.getOriginalData());
					standardRecord.setRequestData(rd.getData());

					standardRecord.setTime(rd.getTime());

					standardRecord.setRecordType(RecordType.BOTH);
					reqMap.remove(rd.getUid());
					mos.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get(),  rd.getServiceId());
					// context.write(new Text(JSON.toJSONString(standardRecord)), null);
				} else {
					rspMap.put(rd.getUid(), rd);
				}
			}

		}
		if (reqMap.size() > 0) {
			for (Record rd : reqMap.values()) {
				standardRecord=new StandardRecord();
				standardRecord.setUid(rd.getUid());
				standardRecord.setServiceId(rd.getServiceId());

				standardRecord.setTime(rd.getTime());
				standardRecord.setRequestData(rd.getData());
				standardRecord.setRequest(rd.getOriginalData());
				standardRecord.setTime(rd.getTime());
				mos.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get(),  rd.getServiceId());
				// context.write(new Text(JSON.toJSONString(standardRecord)), null);
			}
		}

		if (rspMap.size() > 0) {
			for (Record rd : rspMap.values()) {
				standardRecord=new StandardRecord();
				standardRecord.setUid(rd.getUid());
				standardRecord.setServiceId(rd.getServiceId());

				standardRecord.setResponseData(rd.getData());
				standardRecord.setResponse(rd.getOriginalData());
				standardRecord.setTime(rd.getTime());
				standardRecord.setResponseCode(rd.getResponseCode());
				mos.write(new Text(JSON.toJSONString(standardRecord)), NullWritable.get(),  rd.getServiceId());
				// context.write(new Text(JSON.toJSONString(standardRecord)), null);
			}
		}

		if (logger.isDebugEnabled()) {
			logger.debug("reduce(Text, Iterable<Text>, Context) - end");
		}
	}
}
